package com.bicon.botu.finance.wallet.component;

import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.mapper.EntityWrapper;
import com.bicon.botu.core.base.common.utils.EncryptionUtil;
import com.bicon.botu.finance.wallet.common.FinanceConstants;
import com.bicon.botu.finance.wallet.common.WalletConstants;
import com.bicon.botu.finance.wallet.entity.Wallet;
import com.bicon.botu.finance.wallet.entity.WalletDetail;
import com.bicon.botu.finance.wallet.handler.WalletHandler;
import com.bicon.botu.finance.wallet.service.WalletDetailService;
import com.bicon.botu.finance.wallet.service.WalletService;
import com.bicon.botu.mall.component.rpc.service.InformationRpcService;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.math.BigDecimal;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;

/**
 * @author 孙永刚
 * @date 2018/1/10 17:38
 */
@Slf4j
@Component
public class KafkaTask {

    @Autowired
    private KafkaTemplate kafkaTemplate;

    @Autowired
    private WalletDetailService walletDetailService;

    @Autowired
    private WalletHandler walletHandler;

    @Autowired
    private InformationRpcService informationRpc;

    @Autowired
    private WalletService walletService;

    /**
     * 处理kafka监听到的消息并做相应处理
     *
     * @param record
     */
    @KafkaListener(topics = {"BOTU"})
    public void listen(ConsumerRecord<?, ?> record) {
        log.info("【kafka监听成功：】");
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
        if (kafkaMessage.isPresent()) {
            Object message = kafkaMessage.get();
            JSONObject jsonObject = JSONObject.parseObject(message.toString());
            log.info("kafka返回结果" + jsonObject);
            String result = jsonObject.getString("result");
            String orderNo = jsonObject.getString("out_trade_no");
            log.info("【返回结果result：】" + result);
            log.info("【返回订单号orderNo：】" + orderNo);
            if ("WxPay".equals(jsonObject.getString("type"))) {
                if (!result.equals("fail")) {
                    result = WalletConstants.RESPONSE_MESSAGE_200;
                }
            }
            kafkaMessageSend(orderNo, result);
        }
    }

    /**
     * 将支付结果分发到各个系统
     *
     * @param orderNo 订单号
     * @param result  支付结果标识
     */
    @Transactional
    public void kafkaMessageSend(String orderNo, String result) {
        List<WalletDetail> walletDetails = walletDetailService.selectList(new EntityWrapper<WalletDetail>().eq("pay_order_no", orderNo));
        Map<String, String> map = new HashMap<>();
        map.put("result", result);
        map.put("orderNo", orderNo);
        if (walletDetails.size() > 0) {
            WalletDetail walletDetail = walletDetails.get(0);
            String key = walletDetail.getAppId() + "-" + walletDetail.getModule() + "-" + walletDetail.getBusiness();
            if (!result.equals("fail")) {
                //取消充值成功系统消息保存
               /* if (walletDetail.getSubjects().equals(WalletConstants.WALLET_SUBJECTS_RECAHRGE)) {
                    Map<String, Object> map1 = walletService.messageInfo(walletDetail.getSubjects(), walletDetail.getBusiness(), walletDetail.getUserId(), walletDetail.getMoney().doubleValue(), walletDetail.getFinancialType(), orderNo, "1", "2");
                    informationRpc.messageSave(walletDetail.getAppId(), "finance", "transaction", walletDetail.getUserId(), "金融-交易信息", JSONObject.toJSONString(map1), "您成功充值了￥" + walletDetail.getMoney() + "元", "1");
                } else {
                    Map<String, Object> map2 = walletService.messageInfo(walletDetail.getSubjects(), walletDetail.getBusiness(), walletDetail.getUserId(), walletDetail.getMoney().doubleValue(), walletDetail.getFinancialType(), orderNo, "1", "1");
                    informationRpc.messageSave(walletDetail.getAppId(), "finance", "transaction", walletDetail.getUserId(), "金融-交易信息", JSONObject.toJSONString(map2), "您成功支付了￥" + walletDetail.getMoney() + "元", "1");
                }*/
                if (!walletDetail.getSubjects().equals(FinanceConstants.WALLET_SUBJECTS_RECAHRGE)) {
                    Map<String, Object> map2 = walletService.messageInfo(walletDetail.getSubjects(), walletDetail.getBusiness(), walletDetail.getUserId(), walletDetail.getMoney().doubleValue(), walletDetail.getFinancialType(), orderNo, "1", "1");
                    informationRpc.save(walletDetail.getAppId(), "finance", "transaction", walletDetail.getUserId(), "金融-交易信息", JSONObject.toJSONString(map2), "您成功支付了￥" + walletDetail.getMoney() + "元", "1");
                }
                walletDetail.setState(WalletConstants.WALLET_STATE_SUCCESS);
                boolean state = walletDetail.updateById();
                if (state) {
                    //如果是充值业务
                    if ("finance".equals(walletDetail.getModule()) && "recahrge".equals(walletDetail.getBusiness())) {
                        //修改【转账人】db中的余额
                        Wallet wallet = new Wallet().selectById(walletDetail.getWalletId());
                        BigDecimal totalMoney = wallet.getMoney().add(walletDetail.getMoney());
                        log.info("充值金额为" + walletDetail.getMoney() + "账户可用余额为" + wallet.getMoney() + "总金额为" + totalMoney);
                        wallet.setMoney(totalMoney);
                        wallet.setMoneyMD5(EncryptionUtil.getMD5Str(totalMoney.toString()));
                        wallet.preUpdate();
                        boolean resultR = wallet.updateById();
                        log.info("更新账户金额:" + resultR);

                        walletHandler.cacheBalance(walletDetail.getUserId(), totalMoney);
                    } else {
                        log.info("wallt分发支付通知key: " + key.toLowerCase());
                        kafkaTemplate.send(key.toLowerCase(), JSONObject.toJSONString(map));
                    }

                }
            } else {
                walletDetail.setState(WalletConstants.WALLET_STATE_FAIL);
                boolean state = walletDetail.updateById();
                if (state) {
                    log.info("wallt分发支付通知key: " + key.toLowerCase());
                    kafkaTemplate.send(key.toLowerCase(), JSONObject.toJSONString(map));
                }
            }
        }
    }
}


